#!coding=utf-8
import psycopg2
import json
from kafka import KafkaProducer
from kafka.errors import KafkaError
# from django.core.management.base import BaseCommand
# from deepanalysis.settings import *


# class Command(BaseCommand):
#     help = '账号,手机号关系数据入Kafka'
# 
#     def handle(self, *args, **options):
#         main()


# 连接数据库，获取数据：
def PostGreSQLConnect(database, user, password, host, port, sql):
    try:
        conn = psycopg2.connect(
            database=database,
            user=user,
            password=password,
            host=host,
            port=port
        )
    except Exception as e:
        print("连接数据库失败，%s" % e)
    if conn:
        cur = conn.cursor()
        cur.execute(sql)
        results = cur.fetchall()
        cur.close()
        conn.close()
        print(results)
        return results


def make_data():
    result = []
    for i in range(10000):
        data = (1, 'wx_id' + str(i), 'nick_name' + str(i),'chat_count'+str(i), "1", 'phone' + str(i), True, True)
        result.append(data)
    return result

# 数据库数据转指定json格式：
def ResultHandle(results):
    handleResult = []
    for result in results:
        ##添加relation关系
        relations = []
        is_login = result[6]
        is_register = result[7]
        if is_login:
            relations.append("login")
        if is_register:
            relations.append("register")
        # 每种relation作为一种关系，构建node1-rel-node2
        for relation in relations:
            # node1
            type = str(result[0])
            wx_id = result[1]
            nick_name = result[2]
            account = result[3]
            case_involved = str(result[4])
            objectType1 = "account"
            temp_dict1 = {"type": type, "wx_id": wx_id, "nick_name": nick_name, "account": account,
                          "case_involved": str(case_involved), "objectType": objectType1}
            # node2
            involved = "1"
            # if result[5] in [None, '']:
            #     continue
            # else:
            phone = result[5]
            objectType2 = "phone"
            temp_dict2 = {"involved": involved, "phone": phone, "objectType": objectType2}
            # rel
            label = relation
            count = "1"
            temp_dict3 = {"label": label, "count": count}
            # node1,node2,rel
            temp_dict = {"node1": temp_dict1, "node2": temp_dict2, "relation": temp_dict3}
            json_result = json.dumps(temp_dict, ensure_ascii=False)
            handleResult.append(json_result)
    return handleResult


# kafka生产者
def CreateKafkaProducer(kafka_address):
    try:
        producer = KafkaProducer(bootstrap_servers=kafka_address)
        return producer
    except KafkaError as e:
        print(e)


# 发送消息
def SendData(data, producer, kafkatopic):
    try:
        # producer.send(kafkatopic, data.encode('utf-8'))
        producer.send(kafkatopic, data)
        producer.flush()
        # producer.close()
    except KafkaError as e:
        print('KafkaError=========', e)


def main():
    ##配置参数信息：

    # postgres_host = DATABASES['default']['HOST']  # 数据库地址
    # # 数据库端口,setting中获取不到给默认端口'5432'
    # if DATABASES['default'].has_key("PORT"):
    #     postgres_port = DATABASES['default']['PORT']
    # else:
    #     postgres_port = '5432'
    # postgres_user = DATABASES['default']['USER']  # 数据库用户名
    # postgres_password = DATABASES['default']['PASSWORD']  # 数据库密码
    # postgres_database = DATABASES['default']['NAME']  # 数据库名字
    # kafka_host_port = bootstrap_servers  # kafka地址:端口
    # kafka_topic = 'case-search'

    postgres_host = "192.168.30.247"  # 数据库地址
    postgres_port = "5432"  # 数据库端口
    postgres_user = "postgres"  # 数据库用户名
    postgres_password = "shukun123"  # 数据库密码
    postgres_database = "chatdb4"  # 数据库名字
    kafka_host_port = "192.168.12.130"  # kafka地址
    kafka_port = "9092"  # kafka端口
    kafka_topic = 'mytest3'



    sql_command1 = '''
        SELECT
    	ac.type,
    	ac.wx_id,
    	ac.nick_name,
    	ac.chat_account,
    	ac.case_involved,
    	ph.phone_number,
    	rel.is_register,
    	rel.is_login 
        FROM
    	    ( SELECT DISTINCT is_register, is_login, accountbase_id, phonebase_id FROM baseinfo_accountphonerelation WHERE is_register = TRUE OR is_login = TRUE ) AS rel
    	    INNER JOIN baseinfo_accountbasinfo ac ON ac.id = rel.accountbase_id
    	    INNER JOIN ( SELECT phone_number, id FROM baseinfo_phonebaseinfo WHERE phone_number IS NOT NULL AND phone_number != '' ) AS ph ON ph.id = rel.phonebase_id'''


    # database_result = PostGreSQLConnect(postgres_database, postgres_user, postgres_password, postgres_host,
    #                                     postgres_port, sql_command1)
    database_result=make_data()

    handle_result = ResultHandle(database_result)
    producer = CreateKafkaProducer(kafka_host_port)
    for single in handle_result:
        SendData(single, producer, kafka_topic)


if __name__ == '__main__':
    main()